package com.nami.message.controller.middleware.mq.consumer;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import com.google.protobuf.InvalidProtocolBufferException;
import com.nami.gate.external.basic.constants.StoreTopicConst;
import com.nami.im.network.model.message.v2.Message;
import com.nami.message.facade.pmsg.PrivateMsgAddFacade;
import com.nami.message.facade.pmsg.request.PrivateMsgAddCmd;
import com.nami.transport.proto.codec.ImMessages;
import com.nami.transport.proto.codec.MqTransMsg;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.dromara.common.core.utils.ConverUtils;
import org.dromara.common.json.utils.JsonUtils;
import org.springframework.stereotype.Component;

import java.util.Date;

@Slf4j
@Component
@AllArgsConstructor
@RocketMQMessageListener(topic = StoreTopicConst.PCHAT_STORE_MSG,
    consumerGroup = StoreTopicConst.PCHAT_STORE_MSG_CONSUMER_GROUP,
    consumeMode = ConsumeMode.CONCURRENTLY)
public class PrivateMsgConsumer implements RocketMQListener<MessageExt> {

    private final PrivateMsgAddFacade privateMsgAddFacade;

    @Override
    public void onMessage(MessageExt msg) {
        try {
            TimeInterval timer = DateUtil.timer();
            PrivateMsgAddCmd pChatMsgCmd = parsePrivateMsg(msg.getBody());
            privateMsgAddFacade.execute(pChatMsgCmd);

            if (timer.interval() > 200) {
                log.warn("[message] clientMsgId={}, msgId={} 单聊写入消耗：{}ms ", pChatMsgCmd.getClientMsgId(),
                    pChatMsgCmd.getMessageId(), timer.interval());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    private PrivateMsgAddCmd parsePrivateMsg(byte[] msg) throws InvalidProtocolBufferException {
        MqTransMsg.Logic2DispatchBody body = MqTransMsg.Logic2DispatchBody.parseFrom(msg);

        Message downMsg = JsonUtils.parseObject(body.getMessage().toByteArray(), Message.class);
        MqTransMsg.TransUser sender = body.getSender();
        PrivateMsgAddCmd privateMsgAddCmd = new PrivateMsgAddCmd();
        privateMsgAddCmd.setClientMsgId(downMsg.getMsgId());
        privateMsgAddCmd.setPayload(downMsg.getMsgContent());
        privateMsgAddCmd.setPayloadType(downMsg.getMsgType());
        privateMsgAddCmd.setSendTime(new Date(downMsg.getMsgTime()));
        privateMsgAddCmd.setAppKey(sender.getAppKey());
        privateMsgAddCmd.setMessageId(downMsg.getMsgId());
        privateMsgAddCmd.setSeq(downMsg.getMsgSeqNo());
        privateMsgAddCmd.setReceiverId(downMsg.getTargetId());
        privateMsgAddCmd.setSenderId(downMsg.getSenderId());
        privateMsgAddCmd.setCid(ConverUtils.generateCid(downMsg.getSenderId(), downMsg.getTargetId(), false));

        return privateMsgAddCmd;
    }


}
